﻿#include <iostream>
#define SYNC_BOOST_SOCKET_TCP
#ifdef SYNC_BOOST_SOCKET_TCP
#include "ring_buffer.h"

#include <boost/asio.hpp>
#include <boost/thread.hpp>
#include <boost/shared_ptr.hpp>
#include "server.h"

#define DATA_LEN (1<<22) //4MB
#define CACHE_LEN (1 << 29) //512MB


#ifdef _WIN32
typedef struct
{
	uint8_t data[DATA_LEN - 4];
	unsigned int	count;
}pket;
#else
typedef struct
{
	uint8_t data[DATA_LEN - 4];
	unsigned int	count;
}__attribute__((packed)) pket;

#endif
// (__attribute__ package)pket;

using namespace std;
using namespace boost::asio;


bool sockValid = false;
ring_buffer rbuf;
ring_buffer sbuf;
// 所有asio类都需要io_service对象
io_service iosev;
ip::tcp::socket client(iosev);

bool aceptThrdExit = false;
bool txThrdExit = false;
bool rxThrdExit = false;
#ifdef OS_LINUX
pthread_t sid;
pthread_t rid;
pthread_t sigid;
#endif

void reconnect();

void* recvRoutin(void *handle)
{
	//if(!handle)
	//	return;
	//Server *p = (xServer*)handle;
	cout << "start Recv service.." << endl;
	boost::system::error_code ec;
	while (!rxThrdExit)
	{
		//char buf[12] = {0x48,49,0x50,0x51,0x52,0x53,0x54,0x55,0x56,0x57,0x58,0x59};
		if (sockValid)
		{
			uint8_t buf2[13];
			int len1 = read(client, buffer(buf2), transfer_exactly(13), ec);
#ifdef __DEBUF
			cout << "server recv  len = " << len1 << endl;
#endif
			if (len1 == 0)
			{
				sockValid = false;
				client.close();
			    cout << __func__ << ":" << __LINE__ << "Link down!!!" << endl;
			}
			else
			{
				// int count = ring_buffer_put(&rbuf, buf2, len1);
#ifdef __DEBUG
				cout << "save to buf " << count << "byte:" << buf2 << endl;
#endif
			}
		}
		else
		{
			// cout << __func__ << ":" << __LINE__ << "Link down!!!" << endl;
#ifdef __OS_LINUX
			usleep(1000*1000);
#else
			Sleep(500);
#endif
		}
	}
	cout << "rxRoutin Exit normal!!" << endl;
	return NULL;
}


void* sendRoutin(void *handle)
{
	//if(!handle)
	//	return;
	//Server *p = (xServer*)handle;
	cout << "start Send service.." << endl;
	boost::system::error_code ec;
	while (!txThrdExit)
	{
		//char buf[12] = {0x48,49,0x50,0x51,0x52,0x53,0x54,0x55,0x56,0x57,0x58,0x59};
		if (sockValid)
		{
			if (ring_buffer_len(&sbuf) >= DATA_LEN)
			{
				static uint8_t buf1[DATA_LEN];
				ring_buffer_get(&sbuf, buf1, DATA_LEN);
				unsigned int len = ((pket*)buf1)->count;
				if(len == 0)
					continue;

				int len1 = write(client, buffer(buf1), transfer_exactly(len), ec);
				//int len1 = client.write_some(buffer(buf1));
				if (len1 == 0)
				{
					sockValid = false;
					client.close();
					cout << __func__ << ":" << __LINE__ << "Link down!!!" << endl;
				}
			}
		}
		else
		{
			// cout <<__func__<<":"<<__LINE__<< "Link down!!!" << endl;
#ifdef __OS_LINUX
			usleep(500);
#else
			Sleep(500);
#endif
		}
	}
	cout << "txRoutin Exit normal!!" << endl;
	return NULL;
}

#ifdef USE_IGNORE_SIGPIPE 
void signal_handler(const boost::system::error_code& err, int signal)
{
	switch (signal) {
	case SIGPIPE:
		std::cout << "SIGNINT" << std::endl;
		break;
	case SIGTERM:
		std::cout << "SIGNTERM" << std::endl;
		break;
	default:
		break;
	}
}

#endif

void* Accept(void* p)
{
	short port = *((short*)p);
	while(!aceptThrdExit)
	{ 
		if (!sockValid)
		{
			ip::tcp::acceptor acceptor(iosev,
				ip::tcp::endpoint(ip::tcp::v4(), port));
			acceptor.accept(client);
			sockValid = true;
			cout << __func__ << ":" << __LINE__ << "客户端连接成功!!!" << endl;
		}
#ifdef __OS_LINUX
		sleep(1);
#else
		Sleep(1000);
#endif
	}
	cout << "aceptThread Exit normal!!" << endl;
	
	return NULL;
}

int server::Send(uint8_t* buf, int len)
{
	if (len > DATA_LEN)
	{
		cout << "package data over 1MB!!!" << endl;
		return -1;
	}

	pket p;
	memcpy(p.data, buf, len);
	p.count = len;
	cout << __LINE__ << ": "<< p.count <<endl;
	return ring_buffer_put(&sbuf, (uint8_t*)&p, DATA_LEN);
}

int Recv(uint8_t* buf, int len)
{
	int len1 = ring_buffer_len(&rbuf);
#ifdef __DEBUG
	cout << "rbuf has " << len1 << "data" << endl;
#endif
	if (len1 >= len)
		return ring_buffer_get(&rbuf, buf, len);
	else
	{
		cout << "no data" << endl;
		return 0;
	}
}

boost::thread *tThrd = new boost::thread();
boost::thread *rThrd = new boost::thread();
boost::thread *aThrd = new boost::thread();


int initServer(short lport)
{
	//server socket prepare and waiting for connect
	aThrd = new boost::thread(&Accept, &lport);
	//memcpy(aThrd, &acptThrd, sizeof(boost::thread));

	//1. init socket sendbuf(4MB)
	char* buf = (char*)calloc(1, CACHE_LEN);
	if (buf)
		ring_buffer_init(&sbuf, buf, CACHE_LEN);
	else
		return -1;

	char* mem = (char*)calloc(1, CACHE_LEN);
	if (mem)
		ring_buffer_init(&rbuf, mem, CACHE_LEN);
	else
	{
		free(buf);
		return -1;
	}

	//2.start send engin
	tThrd = new boost::thread(&sendRoutin, &sbuf);
	//memcpy(tThrd, &txThrd, sizeof(boost::thread));
#ifdef OS_LINUX
	pthread_create(&sid, NULL, sendRoutin, &sbuf);
#endif
	//3.start recieve engin
	rThrd= new	boost::thread(&recvRoutin, &rbuf);
	//memcpy(rThrd, &rxThrd, sizeof(boost::thread));

	return 0;
}

#ifdef USE_IGNORE_SIGPIPE
void* registerSig(void* p)
{
	cout << "register sig pipe.." << endl;
	signal_set sigset(iosev, SIGPIPE, SIGTERM);
	sigset.async_wait(signal_handler);
	boost::system::error_code ec;
	iosev.run(ec);
}

#endif 


#endif
#ifdef __MAIN__
int lPort = 11002;
int main(int argc, char* argv[])
{
	int ret = initServer();
	if (ret < 0)
	{
		cout << "init server error, check..." << endl;
		return -1;
	}

	//send test data
	uint8_t buf1[12] = { 0x48,49,0x50,0x51,0x52,0x53,0x54,0x55,0x56,0x57,0x58,0x59 };

	uint8_t recvData[13];
	while (1)
	{
		if (sockValid)
		{
			int scount = Send(buf1, sizeof(buf1));
			//cout<<"server send count:"<<scount;
			Sleep(1000);

			int rcount = Recv(recvData, 13);
			if (rcount > 0)
				cout << recvData << endl;
		}
		else
			Accept(lPort);
		
	}

	return 0;
}
#endif

void destroyServer()
{
	aceptThrdExit = true;
	rxThrdExit = true;
	txThrdExit = true;
#ifdef __OS_LINUX
	sleep(2);
#else
	Sleep(2000);
#endif
	//2.stop recvRoutin
	rThrd->join();
	delete  rThrd;
	//3.stop sendRoutin
	tThrd->join();
	delete tThrd;
	//4.release sendbuf
	free(rbuf.buffer);
	//5.release recvbuf
	free(sbuf.buffer);

	//1.stop acptThread
	aThrd->join();
	delete aThrd;

}
server::server(short lport)
{
	int ret = initServer(lport);
	if (ret < 0)
		cout << "init server error, check..." << endl;
}


server::~server()
{
	destroyServer();
}
